package com.dwm;

import com.alibaba.fastjson.JSONObject;
import com.utils.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;

/**
 * 在 DWM 层中，使用双流 join 实现事实表和事实表的拉宽，将订单数据和用户数据关联起来，并在输出流中发送到相应的 Kafka 主题中
 */
public class Seven {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //从kafka上读取到order_info和user_info表的数据
        DataStreamSource<String> orderInfoDs = env.addSource(KafkaUtils.createConsumer("dwd_orderInfo_db", "hwusw"));
        DataStreamSource<String> userInfoDs = env.addSource(KafkaUtils.createConsumer("dwd_userInfo_db", "sshwusw"));
        //形成宽表
        //1.先设置水位线
        SimpleDateFormat smt=new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");

        SingleOutputStreamOperator<JSONObject> orderInfoStream = orderInfoDs.map(x -> {
            return JSONObject.parseObject(x);
        }).filter(x->{
            String string = x.getJSONObject("after").getString("create_time");
            if(string==null || string.length()==0){
                return false;
            }
            return true;
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {
                        try {
                            String ts = jsonObject.getJSONObject("after").getString("create_time");

                            return smt.parse(ts).getTime();
                        } catch (ParseException e) {
                            throw new RuntimeException(e);
                        }

                    }
                }));

        SingleOutputStreamOperator<JSONObject> userInfoStream = userInfoDs.map(x -> {
            return JSONObject.parseObject(x);
        }).filter(x->{
                    String string = x.getJSONObject("after").getString("operate_time");
                    if(string==null || string.length()==0){
                        return false;
                    }
                    return true;
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject jsonObject, long l) {
                        try {
                            String ts = jsonObject.getJSONObject("after").getString("operate_time");
                            return smt.parse(ts).getTime();
                        } catch (ParseException e) {
                            throw new RuntimeException(e);
                        }
                    }
                }));
        //开始关联
        SingleOutputStreamOperator<JSONObject> wideDs = orderInfoStream.keyBy(x -> {
                    return x.getJSONObject("after").getString("user_id");
                }).intervalJoin(userInfoStream.keyBy(x -> {
                    return x.getJSONObject("after").getString("id");
                })).between(Time.hours(-2), Time.hours(1))
                .process(new ProcessJoinFunction<JSONObject, JSONObject, JSONObject>() {
                    @Override
                    public void processElement(JSONObject jsonObject, JSONObject jsonObject2, ProcessJoinFunction<JSONObject, JSONObject, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                        JSONObject jo = new JSONObject();
                        jo.put("order_info", jsonObject);
                        jo.put("user_info", jsonObject2);
                        collector.collect(jo);
                    }
                });
        wideDs.print();
        //并在输出流中发送到相应的 Kafka 主题中
        wideDs.map(x->{
           return x.toString();
        }).addSink(KafkaUtils.createProduer("dwm_wide_db"));

        env.execute();
    }
}
